package com.shujia.flink.state;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo07FraudDetection {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("master:9092,node1:9092,node2:9092")
                .setTopics("transaction")
                .setGroupId("flink_grp_01")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // KafkaSource默认的并行度 等于分区数量的
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        kafkaDS
                .map(line -> {
                    String[] split = line.split(",");
                    Transaction transaction = new Transaction(split[0], Double.parseDouble(split[1]));
                    return transaction;
                })
                .filter(trans -> trans.amount > 0)
                .keyBy(trans -> trans.id)
                .process(new KeyedProcessFunction<String, Transaction, String>() {

                    Double minAmount = 1.0;
                    Double maxAmount = 500.0;
                    ValueState<Double> lastTransAmount;

                    // 任务启动后会执行一次
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastTransAmount = getRuntimeContext().getState(new ValueStateDescriptor<Double>("lastTransAmount", Types.DOUBLE));
                    }

                    // 每条记录会执行一次
                    @Override
                    public void processElement(Transaction trans, KeyedProcessFunction<String, Transaction, String>.Context ctx, Collector<String> out) throws Exception {
                        Double currentTransAmount = trans.amount;

                        if (lastTransAmount.value() != null) {
                            if (lastTransAmount.value() < minAmount && currentTransAmount > maxAmount) {
                                // 存在欺诈检测行为，输出警告
                                out.collect("当前账户：" + trans.id + "可能存在欺诈风险，上一笔交易的记录为：" + lastTransAmount.value() + "，当前交易记录为：" + currentTransAmount);
                            }
                        }
                        // 更新上一笔交易的ValueState
                        lastTransAmount.update(currentTransAmount);

                    }
                }).print();
        ;

        env.execute();


    }
}

class Transaction {
    String id;
    Double amount;

    public Transaction(String id, Double amount) {
        this.id = id;
        this.amount = amount;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public Double getAmount() {
        return amount;
    }

    public void setAmount(Double amount) {
        this.amount = amount;
    }

    @Override
    public String toString() {
        return "Transaction{" +
                "id='" + id + '\'' +
                ", amount=" + amount +
                '}';
    }
}